package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Code24SortBy {
  def main(args: Array[String]): Unit = {

    /**
     * 统计学生总分，并对总分进行排序求各班级各年龄段的前10数据
     */
    /**
     * 排序算子：
     *    sortBy: 可以指定排序规则，默认是升序排序，如果对于数值类型进行倒排，那么可以添加 -实现
     *    sortByKey: 必须使用KeyValue型RDD 可以对Key进行排序操作 相比较而言，灵活性不够
     *
     */

    val sc = new SparkContext(new SparkConf().setMaster("local").setAppName("SortBy"))

    val stuInfoRDD: RDD[(String, (String, String, String, String))] = sc
      .textFile("scala_code/data/students.txt")
      .map {
        case oneLine => {
          val splitRes: Array[String] = oneLine.split(",")
          (splitRes(0), (splitRes(1), splitRes(2), splitRes(3), splitRes(4)))
        }
      }


    sc
      .textFile("spark_code/data/score.txt", 4)
      .map {
        case oneLine => {
          val splitRes: Array[String] = oneLine.split(",")
          (splitRes(0), splitRes(2).toInt)
        }
      }
      .groupBy(_._1)
      //      .mapValues{
      //        case iterator => {
      //          iterator.map(_._2).sum
      //        }
      //      }
      .mapValues(_.map(_._2).sum)
      .join(
        stuInfoRDD
      )
      .groupBy {
        case (id, (totalScore, (name, age, gender, clazz))) => {
          (clazz, age.toInt)
        }
      }
      .mapValues {
        // iterator中的数据是相同 班级相同年龄的数据
        case iterator => {
          // iterator表是为一个迭代器，是按照指针方式进行取值
          //  0 3 2 4
          //          |
          val list: List[(String, (Int, (String, String, String, String)))] = iterator
            .toList
          list
            .sortBy {
              case (id, (totalScore, (name, age, gender, clazz))) => {
                // 是否需要再给  clazz  age
                -totalScore
              }
            }
            .take(10)
        }
      }
      .flatMap(_._2)
      .map {
        case (id, (totalScore, (name, age, gender, clazz))) => {
          ((clazz, age, totalScore), (name, gender ,id))
        }
      }

      // ascending表示降序排序
      .sortByKey(ascending=false)
      .foreach(println)

    //      .sortBy {
    //        case (id, (totalScore, (name, age, gender, clazz))) => {
    //          (clazz, -age.toInt, -totalScore)
    //        }
    //      }
    //      .foreach(println)


    //      }.foreach(println)


  }
}
